package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.commons.codec.binary.Base64;
import org.apache.commons.math3.distribution.NormalDistribution;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.hbase.HBaseDao;
import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.kafka.KafkaProperties;
import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.SlidingWindow;
import com.aotain.apollo.utils.Constants;
import com.aotain.storm.AbstractBolt;
import com.aotain.storm.TupleHelpers;

public class WindowCalBolt extends AbstractBolt{

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = 1L;

	private int windowLengthInSeconds;
	private int emitFrequencyInSeconds;
	private HBaseDao dao = null;

	private HashMap<String,Double> dipPInfo = new HashMap<String,Double>();//各个目标Ip的熵值
	private Map<String,Integer> tripleMap = new HashMap<String,Integer>();//各个元组的PV
	private Map<String,Integer> dipPVMap = new HashMap<String,Integer>();//目标IP PV数
	private Map<String,Set<String>> dipToTriple = new HashMap<String,Set<String>>();//目标ip下的三元组映射
//	private Set<String> IPSet = new HashSet<String>();//要过滤的ip
	private Map<String,Integer> sdPVMap = new HashMap<String,Integer>();//各个源ip到目标ip的PV数

	String dip = null;
	String dport = null;
	String sip = null;
	String sport = null;
	byte[] gisbytes =  null;
	String gisstr = null;
	String tripleKey = null;
	String sdipKey = null;  
	String sourceAreaId = null;
	String sourceAreaCountry = null;
	String destAreaName = null;
	String destGis = null;
	String sourceAreaName = null;
	String sourceGis = null;
	String areaName = null;

	private SlidingWindow cache;

	public WindowCalBolt() {
		this(Constants.WindowLengthInSeconds,Constants.EmitFrequencyInSeconds);
	}

	//	public WindowCalBolt(HashMap<Long,IPDatabase> map,ApolloConfig ap) {
	//		this.map = map;
	//		this.ap = ap;
	//	}

	public WindowCalBolt(int windowLengthInSeconds,int emitFrequencyInSeconds ){
		if(windowLengthInSeconds%emitFrequencyInSeconds!=0){
			System.err.println("窗口个数要是整数");
		}
		this.windowLengthInSeconds = windowLengthInSeconds;
		this.emitFrequencyInSeconds = emitFrequencyInSeconds;
		//		cache = new SlidingWindow(windowLengthInSeconds/emitFrequencyInSeconds);
		cache = new SlidingWindow(20);
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub

	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60);
		return conf;
	}


	//计算需要的值
	private void countInLocal(Tuple tuple) {

		dip = tuple.getString(0);
		dport = tuple.getString(1);
		sip = tuple.getString(2);
		sport = tuple.getString(3);
//		System.err.println("gis64========" + tuple.getString(4));
		gisbytes =  Base64.decodeBase64(tuple.getString(4));
		gisstr = new String(gisbytes);
//		System.err.println("gisdencode ========" + gisstr);
		tripleKey = dip + "_" + dport + "_" + sip + "_" + gisstr;
		sdipKey = dip + "_" + sip;   
  
//						System.err.println("countInlocal====" + tripleKey );
//		if(IPSet.contains(dip)) { 
			int dipPV = 1;
			int triplePV = 1;
			int sdPV = 1;

			if(dipPVMap.get(dip) != null) {
				dipPV = dipPVMap.get(dip) + 1;
			}
			if(tripleMap.get(tripleKey) != null) {
				triplePV = tripleMap.get(tripleKey) + 1;
			}
			if(sdPVMap.get(sdipKey) != null) {
				sdPV = sdPVMap.get(sdipKey) + 1;
			}

			Set<String> tripleSet = dipToTriple.get(dip);
			if(tripleSet == null) {
				tripleSet = new HashSet<String>();
			}
			tripleSet.add(tripleKey);
			dipToTriple.put(dip, tripleSet);

			sdPVMap.put(sdipKey, sdPV);
			dipPVMap.put(dip, dipPV);
			tripleMap.put(tripleKey, triplePV);
//		}

	}

	//定时发送
	private void emitCountingData(OutputCollector collector) {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver,logserver,brokerList);

		//求熵
		for(Entry<String,Set<String>> entry:dipToTriple.entrySet()) {
			String dip = entry.getKey();
			Set<String> tripleSet = entry.getValue();
			//			System.out.println("tripleSet ======" + tripleSet);
			Integer dipPV = dipPVMap.get(dip);
			double pInfo = 0.0;
			for(Iterator<String> iter=tripleSet.iterator();iter.hasNext();) {
				String triple = iter.next();
				//				System.out.println(triple + " :triplePV=====" + tripleMap.get(triple) + "  dipPV=====" + dipPV);
				double p = (double)tripleMap.get(triple)/(double)dipPV;
				double tmpInfo = -Math.log(p)*p;
				pInfo += tmpInfo;
			}
			//			System.out.println(dip + " 's pInfo--->" + pInfo);
			dipPInfo.put(dip, pInfo);
		}
		
		//		System.out.println("dipPInfo ---------->"+dipPInfo);
		cache.addPInfo(dipPInfo);

		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
		
		Map<Integer,HashMap<String,Double>> infoMap = cache.getTuplesWindow();//得到滑动窗口中关于熵的数据

		//		System.out.println("infoMap Size===>" + infoMap.size());
		//		System.out.println("emitNum is "+ emitNum + " and infoMap ===>" + infoMap);

		Map<String,Integer> pInfoLengthMap = new HashMap<String,Integer>();//求熵值平均值时的长度
		Map<String,Double> totalP = new HashMap<String,Double>();//当前所有窗口对应目标元组的熵值和的Map
		Map<String,Double> avgP = new HashMap<String,Double>();//目标元组熵值的均值
		if(infoMap != null ) {
			for(HashMap<String,Double> singleMap: infoMap.values()) {
				if(singleMap != null) {
					for(Entry<String,Double> entry : singleMap.entrySet()) {
						String key = entry.getKey();
						Double value = entry.getValue();
						Double totalPvalue = totalP.get(key);
						if(value != 0) {
							Integer sum = pInfoLengthMap.get(key);
							if(sum == null) {
								sum = 0;
							}
							sum ++;
							pInfoLengthMap.put(key, sum);
							if(totalPvalue == null) {
								totalPvalue = value;
							} else {
								totalPvalue += value;
							}
							totalP.put(key, totalPvalue);
						}
						//						System.out.println(pInfoLengthMap.get(key)+": singleMap---->" + singleMap);
					}
				}
			}
		}

		for(String dkey : totalP.keySet()) {
			double avgPvalue = totalP.get(dkey) / (double) pInfoLengthMap.get(dkey);
			//			System.err.println(dkey + ": totalP --->" + totalP.get(dkey) + " length:" + pInfoLengthMap.get(dkey) + " avgP : " + avgPvalue);
			avgP.put(dkey, avgPvalue);
		}

		HashMap<String,Double> subsqrMap = new HashMap<String,Double>();
		for(String key : avgP.keySet()) {
			double value = 0.0;
			if(dipPInfo.get(key) != null && avgP.get(key)!= null) {
				value = (dipPInfo.get(key)-avgP.get(key))*(dipPInfo.get(key)-avgP.get(key));
			}
			subsqrMap.put(key, value);
		}
		
		cache.addSubsqr(subsqrMap);

		Map<String,Double> totalSubsqr = new HashMap<String,Double>();//差的平方和汇总
		Map<String,Double> stdMap = new HashMap<String,Double>();//标准差集合
		Map<Integer,HashMap<String,Double>> winSums = cache.getSumWindow();

		//		System.out.println("winSums======>" + winSums);

		for(HashMap<String,Double> singleMap: winSums.values()) {
			if(singleMap != null) {
				for(Entry<String,Double> entry : singleMap.entrySet()) {
					String key = entry.getKey();
					Double value = entry.getValue();
					Double totalSvalue = totalSubsqr.get(key);
					if(totalSvalue == null) {
						totalSvalue = 0.0;
					} else {
						totalSvalue += value;
					}
					totalSubsqr.put(key, totalSvalue);
				}
			}
		}

		//公式是：S_n=√(1/(n-1) ∑_(i=1)^n▒〖(x_i-x)〗^2 ) 、S_xn=S_n/√n
		for(String key : totalSubsqr.keySet()) {
			double stdValue = 0.0;
			double sxn = 0.0;
			double confidenceValueMax = 0.0;//最大置信区间
			if(pInfoLengthMap.get(key) != null && pInfoLengthMap.get(key) >1) {
				
				Date date = new Date();
//				String dateStr = sdf.format(date);
				String dateStr_h = sdf_h.format(date);
				String dateStr_d = sdf_d.format(date);
				String dateStr_hour = sdf_hour.format(date);


				stdValue = Math.sqrt(totalSubsqr.get(key)/(pInfoLengthMap.get(key)-1));//S_n
				sxn = stdValue / Math.sqrt(pInfoLengthMap.get(key));//S_xn=S_n/√n
				NormalDistribution s = new NormalDistribution (0,1);//求正态分布分位数所用
				double nordis = s.cumulativeProbability(sxn);
				Double avg = avgP.get(key)==null?0.0:avgP.get(key);
				Double dipP = dipPInfo.get(key)==null?0.0:dipPInfo.get(key);
				confidenceValueMax = avg + nordis;

				Set<String> tripleSet = dipToTriple.get(key);
				String rowKeyAbnormal = "";//dip date sip
				String rowKeyNormal = "";//dip date sip
				String rowKeyAbnSessH = key + "_" + dateStr_h;//异常流量小时表rowkey
				String rowKeyAbnSessD = key + "_" + dateStr_d;//异常流量天表rowkey
				String rowKeyAbnSessSourH = "";//异常流量来源小时表rowkey
				String rowKeyAbnSessSourD = "";//异常流量来源天表rowkey
				String rowKeyAbnArea = "";
				String rowKeyIdcH = dateStr_h;

				//dateStr sip sport dip dport attribution PV
				if(tripleSet != null) {

					for(Iterator<String> iter=tripleSet.iterator();iter.hasNext();) {
						//						System.err.println("log into Hbase --------------->");
						
						String triple = iter.next();
						String sip = triple.split("_")[2];
						String dport = triple.split("_")[1];
						String gis = triple.split("_")[3];
//						System.err.println("gis ==================" + gis);
						String sdIP = key + "_" + sip;
//						rowKeyAbnormal = key + "_" + dateStr + "_" + sip + "_" + dport + "_" + 1;

						rowKeyAbnSessSourH = key + "_" + dateStr_h + "_" + sip;
						rowKeyAbnSessSourD = key + "_" + dateStr_d + "_" + sip;

						String[] gisArray = gis.split("#");
						destAreaName = gisArray[0];
						destGis = gisArray[1];
						sourceAreaName = gisArray[2];
						sourceGis = gisArray[3];
						sourceAreaCountry = gisArray[4];
						sourceAreaId = gisArray[5];
						areaName = gisArray[6];
						
						long time6 = System.currentTimeMillis();
//						if(IPSet.contains(key)) { 
							if(dipP>confidenceValueMax && sxn !=0) {

								//IDC全网统计-小时 SDS_IDC_STAT_H
								//								dao.insert("SDS_IDC_STAT_H", rowKeyIdcH, "cf", "REPORTTIME", dateStr_h);
								//								dao.incrementValue("SDS_IDC_STAT_H", rowKeyIdcH, "cf", "SESSION", 1);
								//								dao.incrementValue("SDS_IDC_STAT_H", rowKeyIdcH, "cf", "PV", 1);

								
								//异常流量归属地日志入库 SDS_ABNORMAL_AREA
									rowKeyAbnArea = key + "_" + dateStr_d + "_" + sourceAreaId;
									hbaseInstance.Add("SDS_ABNORMAL_AREA", rowKeyAbnArea, "cf", "IP", key);
									hbaseInstance.Add("SDS_ABNORMAL_AREA", rowKeyAbnArea, "cf", "REPORTTIME", dateStr_d);
									hbaseInstance.Add("SDS_ABNORMAL_AREA", rowKeyAbnArea, "cf", "AREA", areaName);
									hbaseInstance.incerment("SDS_ABNORMAL_AREA", rowKeyAbnArea, "cf:PV", tripleMap.get(triple));

								//异常日志入库 SDS_ABNORMAL_LOG
								//								abnQuailifers = new String[]{"SOURCEIP","DESTPORT","ACCESSTIME", "ABRNORMAL", "DESTIP", "DESC", "SOURCEAREA", "SOURCEGEO", "DESTAREA", "EVALUATE", "DESTGEO", "ATTNUM"};
								//								abnValues = new String[]{sip, dport, dateStr, "1", key, "异常流量", sourceAreaName, Sourcelon + "," + Sourcelat, destAreaName, "60", Destlon + "," + Destlat, sdPVMap.get(sdIP)+""};
								//								dao.insert("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", abnQuailifers, abnValues);
								//								abnQuailifers = null;
								//								abnValues = null;
							
								Date date1 = new Date();
								String dateStr1 = sdf.format(date1);	
								rowKeyAbnormal = key + "_" + dateStr1 + "_" + sip + "_" + dport + "_" + 1;	
								int attnum = sdPVMap.get(sdIP);
								String evaluate = "";
								String desc = "";
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "SOURCEIP", sip);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESTPORT", dport);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ACCESSTIME", dateStr1);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ABRNORMAL", "1");
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESTIP", key);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "SOURCEAREA", sourceAreaName);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "SOURCEGEO", sourceGis);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "SOURCECOUNTRY", sourceAreaCountry);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESTAREA", destAreaName);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESTGEO", destGis);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ATTNUM", attnum+"");
								if(attnum<=5) {
									evaluate = "100";
									desc = "正常流量";
								} else if(attnum>5 && attnum<=20) {
									evaluate = "60";
									desc = "异常流量";
								} else if(attnum>20) {
									evaluate = "40";
									desc = "异常流量";
								}
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESC", desc);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "EVALUATE", evaluate);
								

								//SDS_ABN_SESS_STAT_H
								//								sessQuailifersH = new String[]{"IP","REPORTTIME","HOUR"};
								//								sessValuesH = new String[]{key,dateStr_h,dateStr_hour};
								//								dao.insert("SDS_ABN_SESS_STAT_H", rowKeyAbnSessH, "cf", sessQuailifersH, sessValuesH);
								//								dao.incrementValue("SDS_ABN_SESS_STAT_H", rowKeyAbnSessH, "cf", "SESSIONMIDDLE", 1);
								
								hbaseInstance.Add("SDS_ABN_SESS_STAT_H", rowKeyAbnSessH, "cf", "IP", key);
								hbaseInstance.Add("SDS_ABN_SESS_STAT_H", rowKeyAbnSessH, "cf", "REPORTTIME", dateStr_h);
								hbaseInstance.Add("SDS_ABN_SESS_STAT_H", rowKeyAbnSessH, "cf", "HOUR", dateStr_hour);
								hbaseInstance.incerment("SDS_ABN_SESS_STAT_H", rowKeyAbnSessH, "cf:SESSIONMIDDLE", tripleMap.get(triple));

								//SDS_ABN_SESS_STAT_D
								//								sessQuailifersD = new String[]{"IP","REPORTTIME"};
								//								sessValuesD = new String[]{key,dateStr_d};
								//								dao.insert("SDS_ABN_SESS_STAT_D", rowKeyAbnSessD, "cf", sessQuailifersD, sessValuesD);
								//								dao.incrementValue("SDS_ABN_SESS_STAT_D", rowKeyAbnSessD, "cf", "SESSIONMIDDLE", 1);
								
								hbaseInstance.Add("SDS_ABN_SESS_STAT_D", rowKeyAbnSessD, "cf", "IP", key);
								hbaseInstance.Add("SDS_ABN_SESS_STAT_D", rowKeyAbnSessD, "cf", "REPORTTIME", dateStr_d);
								hbaseInstance.incerment("SDS_ABN_SESS_STAT_D", rowKeyAbnSessD, "cf:SESSIONMIDDLE", tripleMap.get(triple));

								//SDS_ABN_SESS_SOUR_STAT_H
								//								sourQuailifersH = new String[]{"DESTIP","SOURCEIP","SOURCEAREA"};
								//								sourValuesH = new String[]{key,sip,sourceAreaName};
								//								dao.insert("SDS_ABN_SESS_SOUR_STAT_H", rowKeyAbnSessSourH, "cf", sourQuailifersH, sourValuesH);
								//								dao.incrementValue("SDS_ABN_SESS_SOUR_STAT_H", rowKeyAbnSessSourH, "cf", "PV", tripleMap.get(triple));

								hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_H", rowKeyAbnSessSourH, "cf", "DESTIP", key);
								hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_H", rowKeyAbnSessSourH, "cf", "SOURCEIP", sip);
								hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_H", rowKeyAbnSessSourH, "cf", "SOURCEAREA", sourceAreaName);
								hbaseInstance.incerment("SDS_ABN_SESS_SOUR_STAT_H", rowKeyAbnSessSourH, "cf:PV", tripleMap.get(triple));

								//SDS_ABN_SESS_SOUR_STAT_D
								//								sourQuailifersD = new String[]{"DESTIP","SOURCEIP","SOURCEAREA"};
								//								sourValuesH = new String[]{key,sip,sourceAreaName};
								//								dao.insert("SDS_ABN_SESS_SOUR_STAT_D", rowKeyAbnSessSourD, "cf", sourQuailifersD, sourValuesH);
								//								dao.incrementValue("SDS_ABN_SESS_SOUR_STAT_D", rowKeyAbnSessSourD, "cf", "PV", tripleMap.get(triple));

								hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_D", rowKeyAbnSessSourD, "cf", "DESTIP", key);
								hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_D", rowKeyAbnSessSourD, "cf", "SOURCEIP", sip);
								hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_D", rowKeyAbnSessSourD, "cf", "SOURCEAREA", sourceAreaName);
								hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_D", rowKeyAbnSessSourD, "cf", "REPORTTIME", dateStr_d);
								hbaseInstance.incerment("SDS_ABN_SESS_SOUR_STAT_D", rowKeyAbnSessSourD, "cf:PV", tripleMap.get(triple));

							} else { 
								//正常日志入库SDS_IDC_LOG
								//								norQuailifers = new String[]{"SOURCEIP","DESTIP","DESTPORT","ACCESSTIME", "PV", "SOURCEAREA", "SOURCEGEO", "DESTAREA", "DESTGEO", "EVALUATE","ABRNORMAL","ATTNUM","DESC"};
								//								norValues = new String[]{sip,key,triple.split("-")[1],dateStr,dipPVMap.get(key)+"",sourceAreaName,Sourcelon + "," + Sourcelat, destAreaName, Destlon + "," + Destlat,"100","0",sdPVMap.get(sdIP)+"","正常流量"};
								//								dao.insert("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", norQuailifers, norValues);

								Date date2 = new Date();
								String dateStr2 = sdf.format(date2);	
								rowKeyNormal = key + "_" + dateStr2 + "_" + sip + "_" + dport + "_" + 0;	
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", "SOURCEIP", sip);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", "DESTIP", key);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", "DESTPORT", triple.split("_")[1]);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", "ACCESSTIME", dateStr2);
//								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", "PV", dipPVMap.get(key)+"");
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", "SOURCEAREA", sourceAreaName);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", "SOURCEGEO", sourceGis);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", "DESTAREA", destAreaName);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", "DESTGEO", destGis);
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", "EVALUATE", "100");
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", "ABRNORMAL", "0");
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", "DESC", "正常流量");
								hbaseInstance.Add("SDS_ABNORMAL_LOG", rowKeyNormal, "cf", "ATTNUM", sdPVMap.get(sdIP)+"");

							}
							
					}

				}
			} 
			stdMap.put(key, stdValue);
		}
		tripleMap.clear();//各元组pv清零
		dipPVMap.clear();//dipPV清零
		dipPInfo.clear();
		dipToTriple.clear();
		sdPVMap.clear();
		cache.advanceHead();
		
	}

	@Override
	public void cleanup() {
      
	}



	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			if(TupleHelpers.isTickTuple(tuple)) {
				emitCountingData(collector);  
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			System.err.println(" _windowcalbolt exception"+e.getMessage());
			Logger.getRootLogger().error("windowcalbolt exception====",e);
			collector.fail(tuple);
			throw new FailedException("WindowCalBolt出异常");
		}	
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		String dbJson = stormConf.get("dbjson").toString();
		ApolloConfig ap2 = new ApolloConfig(dbJson);
		logserver = ap2.getMushroomServer();
		zooserver = KafkaProperties.hbase_zkList;
	}
}
